package com.lhd.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lhd.bean.GoodsStatsBean;
import com.lhd.common.utils.DateFormatUtil;
import com.lhd.common.utils.MyClickHouseUtil;
import com.lhd.common.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * 商品统计数据窗口计算
 * 统计访问商品详情页的商品访客数、商品浏览量、加购收藏人数等数据
 */
public class DwsTradeGoodsStatsWindow {

    public static void main(String[] args) throws Exception {

        // TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2.读取Kafka商品详情页主题数据创建流
        String pageTopic = "dwd_traffic_page_log";
        String cartTopic = "dwd_trade_cart_add";
        String favorTopic = "dwd_interaction_favor_add";
        String groupId = "dws_trade_goods_stats_window";

        DataStreamSource<String> pageKafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId));
        DataStreamSource<String> cartKafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(cartTopic, groupId));
        DataStreamSource<String> favorKafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(favorTopic, groupId));

        // TODO 3.处理页面日志数据
        SingleOutputStreamOperator<JSONObject> pageJsonObjDS = pageKafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    //转换为JSON对象
                    JSONObject jsonObject = JSON.parseObject(value);
                    //获取当前页面id
                    String pageId = jsonObject.getJSONObject("page").getString("page_id");
                    //过滤出商品详情页的数据
                    if ("good_detail".equals(pageId)) {
                        out.collect(jsonObject);
                    }
                } catch (Exception e) {
                    // 忽略解析错误的数据
                }
            }
        });

        // TODO 4.处理加购数据
        SingleOutputStreamOperator<JSONObject> cartJsonObjDS = cartKafkaDS.map(JSON::parseObject);

        // TODO 5.处理收藏数据
        SingleOutputStreamOperator<JSONObject> favorJsonObjDS = favorKafkaDS.map(JSON::parseObject);

        // TODO 6.提取事件时间生成Watermark
        SingleOutputStreamOperator<JSONObject> pageJsonObjWithWmDS =
                pageJsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("ts");
                            }
                        }));

        SingleOutputStreamOperator<JSONObject> cartJsonObjWithWmDS =
                cartJsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                String operateTime = element.getString("operate_time");
                                if (operateTime != null) {
                                    return DateFormatUtil.toTs(operateTime, true);
                                } else {
                                    return DateFormatUtil.toTs(element.getString("create_time"), true);
                                }
                            }
                        }));

        SingleOutputStreamOperator<JSONObject> favorJsonObjWithWmDS =
                favorJsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("ts");
                            }
                        }));

        // TODO 7.按照商品ID分组
        KeyedStream<JSONObject, String> pageKeyedStream =
                pageJsonObjWithWmDS.keyBy(json -> json.getJSONObject("page").getString("item"));

        KeyedStream<JSONObject, String> cartKeyedStream =
                cartJsonObjWithWmDS.keyBy(json -> json.getString("sku_id"));

        KeyedStream<JSONObject, String> favorKeyedStream =
                favorJsonObjWithWmDS.keyBy(json -> json.getString("sku_id"));

        // TODO 8.使用状态编程统计商品访客数、浏览量等
        SingleOutputStreamOperator<GoodsStatsBean> pageGoodsStatsDS =
                pageKeyedStream.flatMap(new RichFlatMapFunction<JSONObject, GoodsStatsBean>() {

                    private ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build();

                        ValueStateDescriptor<String> visitDateStateDes =
                                new ValueStateDescriptor<>("visit-date-state", String.class);

                        //设置TTL
                        visitDateStateDes.enableTimeToLive(ttlConfig);

                        lastVisitDateState = getRuntimeContext().getState(visitDateStateDes);
                    }

                    @Override
                    public void flatMap(JSONObject value, Collector<GoodsStatsBean> out) throws Exception {
                        //获取状态数据以及当前数据中的日期
                        Long ts = value.getLong("ts");
                        String curDt = DateFormatUtil.toDate(ts);

                        String lastVisitDt = lastVisitDateState.value();

                        //定义统计数据
                        long uvCt = 0L;  // 访客数
                        long pvCt = 1L;  // 浏览量

                        //如果状态为空或者状态时间与当前时间不同,则为需要的数据
                        if (lastVisitDt == null || !lastVisitDt.equals(curDt)) {
                            uvCt = 1L;
                            lastVisitDateState.update(curDt);
                        }

                        //获取商品信息
                        String skuId = value.getJSONObject("page").getString("item");
                        String skuName = value.getJSONObject("page").getString("item_name");

                        //输出数据
                        GoodsStatsBean goodsStatsBean = new GoodsStatsBean(skuId, skuName, uvCt, pvCt, ts);
                        goodsStatsBean.setCartAddUserCount(0L);
                        goodsStatsBean.setFavorAddUserCount(0L);
                        out.collect(goodsStatsBean);
                    }
                });

        // TODO 9.处理加购数据
        SingleOutputStreamOperator<GoodsStatsBean> cartGoodsStatsDS =
                cartKeyedStream.flatMap(new RichFlatMapFunction<JSONObject, GoodsStatsBean>() {

                    private ValueState<String> lastCartAddState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build();

                        ValueStateDescriptor<String> cartAddStateDes =
                                new ValueStateDescriptor<>("cart-add-date-state", String.class);

                        //设置TTL
                        cartAddStateDes.enableTimeToLive(ttlConfig);

                        lastCartAddState = getRuntimeContext().getState(cartAddStateDes);
                    }

                    @Override
                    public void flatMap(JSONObject value, Collector<GoodsStatsBean> out) throws Exception {
                        String operateTime = value.getString("operate_time");

                        String curDt = null;
                        if (operateTime != null) {
                            curDt = operateTime.split(" ")[0];
                        } else {
                            String createTime = value.getString("create_time");
                            curDt = createTime.split(" ")[0];
                        }

                        String lastCartAddDt = lastCartAddState.value();

                        //如果状态为空或者状态时间与当前时间不同,则为需要的数据
                        if (lastCartAddDt == null || !lastCartAddDt.equals(curDt)) {
                            lastCartAddState.update(curDt);
                            //获取商品信息
                            String skuId = value.getString("sku_id");
                            String skuName = value.getString("sku_name");

                            //输出数据
                            GoodsStatsBean goodsStatsBean = new GoodsStatsBean(skuId, skuName, 0L, 0L, System.currentTimeMillis());
                            goodsStatsBean.setCartAddUserCount(1L);
                            goodsStatsBean.setFavorAddUserCount(0L);
                            out.collect(goodsStatsBean);
                        }
                    }
                });

        // TODO 10.处理收藏数据
        SingleOutputStreamOperator<GoodsStatsBean> favorGoodsStatsDS =
                favorKeyedStream.flatMap(new RichFlatMapFunction<JSONObject, GoodsStatsBean>() {

                    private ValueState<String> lastFavorAddState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build();

                        ValueStateDescriptor<String> favorAddStateDes =
                                new ValueStateDescriptor<>("favor-add-date-state", String.class);

                        //设置TTL
                        favorAddStateDes.enableTimeToLive(ttlConfig);

                        lastFavorAddState = getRuntimeContext().getState(favorAddStateDes);
                    }

                    @Override
                    public void flatMap(JSONObject value, Collector<GoodsStatsBean> out) throws Exception {
                        String createTime = value.getString("create_time");
                        String curDt = createTime.split(" ")[0];

                        String lastFavorAddDt = lastFavorAddState.value();

                        //如果状态为空或者状态时间与当前时间不同,则为需要的数据
                        if (lastFavorAddDt == null || !lastFavorAddDt.equals(curDt)) {
                            lastFavorAddState.update(curDt);
                            //获取商品信息
                            String skuId = value.getString("sku_id");
                            // 收藏数据中没有sku_name，暂时设为空字符串
                            String skuName = value.getString("sku_name");
                            if (skuName == null) {
                                skuName = "";
                            }

                            //输出数据
                            GoodsStatsBean goodsStatsBean = new GoodsStatsBean(skuId, skuName, 0L, 0L, DateFormatUtil.toTs(createTime, true));
                            goodsStatsBean.setCartAddUserCount(0L);
                            goodsStatsBean.setFavorAddUserCount(1L);
                            out.collect(goodsStatsBean);
                        }
                    }
                });

        // TODO 11.合并三个流
        DataStream<GoodsStatsBean> unionDS = pageGoodsStatsDS.union(cartGoodsStatsDS, favorGoodsStatsDS);

        // TODO 12.按照商品ID分组
        KeyedStream<GoodsStatsBean, String> goodsKeyedStream = unionDS.keyBy(GoodsStatsBean::getSkuId);

        // TODO 13.开窗聚合
        SingleOutputStreamOperator<GoodsStatsBean> resultDS = goodsKeyedStream
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<GoodsStatsBean>() {
                    @Override
                    public GoodsStatsBean reduce(GoodsStatsBean value1, GoodsStatsBean value2) throws Exception {
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setCartAddUserCount(value1.getCartAddUserCount() + value2.getCartAddUserCount());
                        value1.setFavorAddUserCount(value1.getFavorAddUserCount() + value2.getFavorAddUserCount());
                        return value1;
                    }
                }, new org.apache.flink.streaming.api.functions.windowing.WindowFunction<GoodsStatsBean, GoodsStatsBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<GoodsStatsBean> values, Collector<GoodsStatsBean> out) throws Exception {
                        //获取数据
                        GoodsStatsBean goodsStatsBean = values.iterator().next();
                        //补充字段
                        goodsStatsBean.setTs(System.currentTimeMillis());
                        goodsStatsBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        goodsStatsBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        //输出数据
                        out.collect(goodsStatsBean);
                    }
                });

        // TODO 14.将数据写出到ClickHouse
        resultDS.print("商品统计数据>>>>>>>>>>>");
        resultDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_trade_goods_stats_window values(?,?,?,?,?,?,?,?,?)"));
        // TODO 15.启动任务
        env.execute("DwsTradeGoodsStatsWindow");
    }
}